#pragma once

#include "../common/Common.h"
#include "../common/Base.h"
#include "../common/Message.h"

namespace rpc
{

    namespace server
    {

        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;

            void onTopicRequest(const BaseConnection::ptr &conn, const TopicRequest::ptr &req)
            {
                TopicOpType type = req->opType();

                DLOG("RpcTopic.h TopicManager onTopicRequest() ...");

                bool ret = true;
                switch (type)
                {
                case TopicOpType::TOPIC_CREATE:
                    topicCreate(conn, req);
                    break;
                case TopicOpType::TOPIC_DELETE:
                    topicRemove(conn, req);
                    break;
                case TopicOpType::TOPIC_CANCEL:
                    topicCancel(conn, req);
                    break;
                case TopicOpType::TOPIC_SUBCRIBE:
                    ret = topicSubscribe(conn, req);
                    break;
                case TopicOpType::TOPIC_PUBLISH:
                    ret = topicPublish(conn, req);
                    break;
                default:
                    return errorResponse(conn, req, RCode::RCODE_INVALID_OPTYPE);
                }

                if (!ret)
                {
                    return errorResponse(conn, req, RCode::RCODE_NOT_FOUND_TOPIC);
                }

                // 响应成功
                topicResponse(conn, req);
            }

            // 一个订阅者在连接断开时的处理---删除其关联的数据
            void onShutdown(const BaseConnection::ptr &conn)
            {
                vector<Topic::ptr> topicVec;
                Subscriber::ptr sub;
                {
                    unique_lock<mutex> lock(_mtx);

                    // 找出当前订阅者
                    auto it = _subscriberMap.find(conn);
                    if (it == _subscriberMap.end())
                    {
                        return;
                    }
                    sub = it->second;

                    // 找出还存在的订阅主题
                    for (auto &topic : sub->topics)
                    {
                        // 如果存在当前主题
                        auto topicIt = _topicMap.find(topic);
                        if (topicIt == _topicMap.end())
                            continue;
                        topicVec.push_back(topicIt->second);
                    }
                    // 删除管理数据
                    _subscriberMap.erase(conn);
                }

                // 从主题中移除当前订阅者
                for (auto &topic : topicVec)
                {
                    topic->removeSubscribe(sub);
                }
            }

        private:
            void errorResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &req, RCode code)
            {

                TopicResponse::ptr resp = MessageFactory::create<TopicResponse>();

                resp->setMtype(MType::RSP_TOPIC);
                resp->setRcode(code);
                resp->setRid(req->rid());

                conn->send(resp);
            }

            void topicResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &req)
            {
                TopicResponse::ptr resp = MessageFactory::create<TopicResponse>();

                resp->setMtype(MType::RSP_TOPIC);
                resp->setRcode(RCode::RCODE_OK);
                resp->setRid(req->rid());

                conn->send(resp);
            }

            void topicCreate(const BaseConnection::ptr &conn, const TopicRequest::ptr &req)
            {
                string topic = req->topic();
                unique_lock<mutex> lock(_mtx);
                DLOG("主题创建成功: %s ", req->topic().c_str());
                _topicMap.emplace(topic, std::make_shared<Topic>(topic));
            }

            void topicCancel(const BaseConnection::ptr &conn, const TopicRequest::ptr &req)
            {
                Topic::ptr topic;
                Subscriber::ptr sub;
                {
                    unique_lock<mutex> lock(_mtx);
                    auto topicIt = _topicMap.find(req->topic());
                    if (topicIt != _topicMap.end())
                    {
                        topic = topicIt->second;
                    }

                    auto subIt = _subscriberMap.find(conn);
                    if (subIt != _subscriberMap.end())
                    {
                        sub = subIt->second;
                    }
                }

                if (sub)
                {
                    sub->removeTopic(req->topic());
                }
                if (topic && sub)
                {
                    topic->removeSubscribe(sub);
                }
            }

            // 删除主题
            void topicRemove(const BaseConnection::ptr &conn, const TopicRequest::ptr &req)
            {
                // 删除主题
                // 从订阅者中移除当前主题
                unordered_set<Subscriber::ptr> subs;
                {
                    unique_lock<mutex> lock(_mtx);
                    auto it = _topicMap.find(req->topic());
                    if (it != _topicMap.end())
                    {
                        subs = it->second->subscribers;
                        _topicMap.erase(it);
                    }
                }

                for (auto &sub : subs)
                {
                    sub->removeTopic(req->topic());
                }
            }

            bool topicPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &req)
            {
                // 找到当前主题，向其中的订阅者发送消息
                Topic::ptr topic;
                {
                    unique_lock<mutex> lock(_mtx);
                    auto it = _topicMap.find(req->topic());
                    if (it == _topicMap.end())
                    {
                        return false;
                    }

                    topic = it->second;
                }

                topic->publishMessage(req);
                return true;
            }

            bool topicSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &req)
            {
                Topic::ptr topic;
                Subscriber::ptr sub;
                {
                    unique_lock<mutex> lock(_mtx);

                    auto topicIt = _topicMap.find(req->topic());
                    if (topicIt == _topicMap.end())
                    {
                        return false;
                    }

                    topic = topicIt->second;

                    auto it = _subscriberMap.find(conn);
                    if (it == _subscriberMap.end())
                    {
                        sub = std::make_shared<Subscriber>(conn);
                        _subscriberMap.emplace(conn, sub);
                    }
                    else
                    {
                        sub = it->second;
                    }
                }

                topic->appendSubscribe(sub);
                sub->appendTopic(topic->topicName);

                return true;
            }

        private:
            struct Subscriber
            {
                using ptr = std::shared_ptr<Subscriber>;

                mutex mtx;
                BaseConnection::ptr conn;
                unordered_set<string> topics; // 订阅的主题

                Subscriber(const BaseConnection::ptr &c)
                    : conn(c) {}

                // 订阅主题
                void appendTopic(const string &topic)
                {
                    unique_lock<mutex> lock(mtx);
                    topics.emplace(topic);
                }

                // 取消订阅
                void removeTopic(const string &topic)
                {
                    unique_lock<mutex> lock(mtx);
                    topics.erase(topic);
                }
            };

            struct Topic
            {
                using ptr = std::shared_ptr<Topic>;
                mutex mtx;
                string topicName;
                unordered_set<Subscriber::ptr> subscribers; // 当前主题的订阅者

                Topic(const string &name)
                    : topicName(name) {}

                // 新增订阅
                void appendSubscribe(const Subscriber::ptr sub)
                {
                    unique_lock<mutex> lock(mtx);
                    subscribers.emplace(sub);
                }

                // 移除订阅
                void removeSubscribe(const Subscriber::ptr sub)
                {
                    unique_lock<mutex> lock(mtx);
                    subscribers.erase(sub);
                }

                // 收到消息发布请求时调用
                void publishMessage(const BaseMessage::ptr &msg)
                {
                    unique_lock<mutex> lock(mtx);
                    for (auto &subscribe : subscribers)
                    {
                        subscribe->conn->send(msg);
                    }
                }
            };

        private:
            mutex _mtx;
            unordered_map<string, Topic::ptr> _topicMap;
            unordered_map<BaseConnection::ptr, Subscriber::ptr> _subscriberMap;
        };

    }; // server

}; // rpc